package com.niit.streaming

import org.apache.spark.SparkConf
import org.apache.spark.rdd.RDD
import org.apache.spark.streaming.{Seconds, StreamingContext}

import scala.collection.mutable

//利用RDD队列的方式模拟生产数据，做统计
object Spark_stream_Queue {

  def main(args: Array[String]): Unit = {
    val sparkConf = new SparkConf().setMaster("local[*]").setAppName("SparkStream")
    val ssc = new StreamingContext(sparkConf,Seconds(5))
    ssc.sparkContext.setLogLevel("ERROR")

    //创建队列
    val rddQueue = new mutable.Queue[RDD[Int]]()
    //创建队列输入流
    val inputStream = ssc.queueStream(rddQueue,oneAtATime = false);
    //处理队列中的RDD数据  -->统计每个数字出现的次数 1 2 3 4 ，1 2 3 4，1 2 3 4
    val mappedStream = inputStream.map( (_,1) )
    val reduceStream =  mappedStream.reduceByKey(_+_)
    //打印
    reduceStream.print
    //启动任务
    ssc.start()
    //循环创建RDD，并存放到队列当中
    for (i <- 1 to 5){
      rddQueue += ssc.sparkContext.makeRDD(1 to 100,10)
      Thread.sleep(2000)
    }
   ssc.awaitTermination()
  }

}
